/*
 *    Copyright (c) 2023 李伟国
 *    moyu is licensed under Mulan PSL v2.
 *    You can use this software according to the terms and conditions of the Mulan PSL v2.
 *    You may obtain a copy of Mulan PSL v2 at:
 *             http://license.coscl.org.cn/MulanPSL2
 *    THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 *    See the Mulan PSL v2 for more details.
 */

package cn.moyu.kettle;

import cn.moyu.entity.KettleTransInfo;
import cn.moyu.entity.KettleTransParameters;
import cn.moyu.entity.vo.KettleTransKeyVo;
import cn.moyu.entity.vo.KettleTransUpdateVo;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.insertupdate.InsertUpdateMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * trans 运行
 */

@Component
public class KettleRunTrans {


    /**
     * 运行一个转换
     *
     * @param transInfo      转换信息
     * @param transParameters trans运行的参数信息
     * @param dataBaseInput  查询的数据源
     * @param dataBaseOutput 保存的数据源
     * @param keyVo          用于查询的字段
     * @param updateVo       用于更新的字段
     * @throws KettleException kettle异常
     */
    public void runTrans(KettleTransInfo transInfo, List<KettleTransParameters> transParameters, DatabaseMeta dataBaseInput, DatabaseMeta dataBaseOutput, KettleTransKeyVo keyVo, KettleTransUpdateVo updateVo) throws KettleException {
        TransMeta transMeta = new TransMeta();
        transMeta.setName("测试");
        // 添加数据连接
        transMeta.addDatabase(dataBaseInput);
        transMeta.addDatabase(dataBaseOutput);

        // 表输入
        TableInputMeta tableInputMeta = new TableInputMeta();
        tableInputMeta.setDatabaseMeta(dataBaseInput);
        tableInputMeta.setSQL(transInfo.getTransSql());
        tableInputMeta.setVariableReplacementActive(true);

        StepMeta inputStep = new StepMeta("startStep", tableInputMeta);
        inputStep.setLocation(50, 50);
        inputStep.setDraw(true);
        transMeta.addStep(inputStep);

        //插入或更新
        InsertUpdateMeta insertUpdateMeta = getInsertUpdate(transInfo, dataBaseOutput, keyVo, updateVo);
        StepMeta insertUpdateStep = new StepMeta("STEP_INSERT_UPDATE", insertUpdateMeta);
        insertUpdateStep.setLocation(150, 50);
        insertUpdateStep.setDraw(true);
        transMeta.addStep(insertUpdateStep);

        //设置步骤直接的关系
        TransHopMeta hop = new TransHopMeta(inputStep, insertUpdateStep);
        transMeta.addTransHop(hop);
        for(KettleTransParameters parameter: transParameters){
            transMeta.setParameterValue(parameter.getParamKey(), parameter.getParamValue());
        }

        //开始执行
        Trans trans = new Trans(transMeta);
        trans.prepareExecution(null);

        trans.startThreads();
        trans.waitUntilFinished();

    }

    private InsertUpdateMeta getInsertUpdate(KettleTransInfo transInfo,DatabaseMeta dataBaseOutput, KettleTransKeyVo keyVo, KettleTransUpdateVo updateVo){
        InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();
        insertUpdateMeta.setDatabaseMeta(dataBaseOutput);
        //设置要更新的表
        insertUpdateMeta.setTableName(transInfo.getTableName());
        //设置用来查询的关键字
        insertUpdateMeta.setKeyLookup(keyVo.getLookUp());
        insertUpdateMeta.setKeyStream(keyVo.getStream());
        insertUpdateMeta.setKeyStream2(keyVo.getStream2());
        insertUpdateMeta.setKeyCondition(keyVo.getCondition());
        //设置更新的字段
        insertUpdateMeta.setUpdateLookup(updateVo.getLookUp());
        insertUpdateMeta.setUpdateStream(updateVo.getStream());
        insertUpdateMeta.setUpdate(updateVo.getIsUpdate());
        return insertUpdateMeta;
    }
}
